package com.zidongxiangxi.rabbit.consumer.interceptor;

import com.zidongxiangxi.rabbit.consumer.manager.ConsumeRecordManager;
import com.zidongxiangxi.rabbit.core.util.RabbitUtils;
import org.aopalliance.intercept.MethodInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.ImmediateRequeueAmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.Date;
import java.util.Objects;

/**
 * 顺序消费拦截器
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class SequenceOperationsInterceptor extends AbstractConsumerOperationsInterceptor {
	private final Logger logger = LoggerFactory.getLogger(SequenceOperationsInterceptor.class);

	private static final int DEFAULT_CONSUME_FAIL_DELAY = 10000;
	private static final String CAN_NOT_CONSUME_YET = "消息还不能被消费，发送应用为：%s，前置消息为：%s";

	/**
	 * 当前应用
	 */
	private final String application;
	/**
	 * 消费记录manager
	 */
	private final ConsumeRecordManager consumeRecordManager;
	/**
	 * 无法顺序消费时，consumeFailDelay毫秒后可以被强行消费
	 */
	private final int consumeFailDelay;

	public SequenceOperationsInterceptor(
			String application,
			ConsumeRecordManager consumeRecordManager,
			int consumeFailDelay
	) {
		this.application = application;
		this.consumeRecordManager = consumeRecordManager;
		if (consumeFailDelay <= 0) {
			this.consumeFailDelay = DEFAULT_CONSUME_FAIL_DELAY;
		} else {
			this.consumeFailDelay = consumeFailDelay;
		}
	}

	@Override
	public Object invoke(MethodInvocation methodInvocation) throws Throwable {
		Message message = getMessage(methodInvocation);
		if (Objects.isNull(message)
				|| Objects.isNull(message.getMessageProperties())
				|| CollectionUtils.isEmpty(message.getMessageProperties().getHeaders())) {
			return methodInvocation.proceed();
		}
		String produceApplication = RabbitUtils.getProduceApplication(message);
		String messageId = RabbitUtils.getMessageId(message);
		if (StringUtils.isEmpty(produceApplication) || StringUtils.isEmpty(messageId)) {
			return methodInvocation.proceed();
		}
		// 不是顺序消息，直接执行
		if (!RabbitUtils.isSequenceMessage(message)) {
			return consumeMessage(methodInvocation, produceApplication, messageId);
		}
		String previousMessageId = RabbitUtils.getPreviousMessageId(message);
		Date msgTimestamp = RabbitUtils.getTimestamp(message);
		// 判断是否已经可以消费
		if (isCanConsume(application, messageId, previousMessageId, msgTimestamp)) {
			return consumeMessage(methodInvocation, produceApplication, messageId);
		}
		// 回队列
		throw new ImmediateRequeueAmqpException(String.format(CAN_NOT_CONSUME_YET, produceApplication,
				previousMessageId));
	}

	private boolean isCanConsume(String produceApplication, String messageId, String previousMessageId,
								 Date msgTimestamp) {
		boolean canConsume = false;
		try {
			canConsume = consumeRecordManager.isConsumed(produceApplication, application, previousMessageId);
		} catch (Throwable e) {
			logger.error("fail to check sequence is can consume, message id: {}", messageId, e);
		}
		if (!canConsume
				&& Objects.nonNull(msgTimestamp)
				&& System.currentTimeMillis() - msgTimestamp.getTime() >= consumeFailDelay) {
			canConsume = true;
		}
		return canConsume;
	}

	private Object consumeMessage(
			MethodInvocation methodInvocation,
			String produceApplication,
			String messageId
	) throws Throwable {
		Object result = methodInvocation.proceed();
		// 成功执行，尝试添加消费记录，catch所有异常，不影响mq的消费
		// 如果开启了幂等消费，则由幂等拦截器添加消费记录，下面语句不会添加成功
		try {
			consumeRecordManager.saveRecord(produceApplication, application, messageId);
		} catch (Throwable e) {
			logger.error("保存消费记录失败", e);
		}
		return result;
	}
}
